package com.surfilter.mq.rocketmq.exe;

import com.surfilter.mq.rocketmq.RocketMqConfig;
import com.surfilter.mq.rocketmq.RocketMqMessageStandardOprate;
import com.surfilter.mq.rocketmq.cache.MesgTaskManage;
import com.surfilter.mq.rocketmq.cache.RockMqMesgManage;
import com.surfilter.mq.rocketmq.cache.RocketMqMessageStandard;
import com.surfilter.mq.rocketmq.timer.ConsumerCurrencyCommon;
import com.surfilter.mq.threadman.ThreadExecuteManager;
import com.surfilter.mq.threadman.ThreadManage;
import com.surfilter.mq.threadman.ThreadManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationTargetException;
import java.util.List;


public class RocketMesgExecute {
	
	private static final Logger logger = LoggerFactory.getLogger(RocketMesgExecute.class);
	
	private RocketMqConfig config;
	
	private ThreadManager threadManage;
	
	private RockMqMesgManage manage;
	
	private Class<? extends RocketMqMessageStandardOprate> oprate;
	
	private RocketMqMessageStandardOprate r_oprate;
	
	private ExecuteType type;
	
	private int threadNum;
	
	private boolean init_type = false;
	
	private ExecuteModel model = ExecuteModel.Nonblocking_queue;
	
	private static boolean executing = false;
	
	private String single_key;
	
	public RocketMesgExecute(String single_key, RocketMqConfig config, RockMqMesgManage manage, int threadNum, ExecuteModel model){
		this.single_key = single_key;
		this.config = config;
		this.manage = manage;
		this.threadNum = threadNum;
		if(model != null)this.model = model;
	}

	public RocketMesgExecute(String single_key, RocketMqConfig config, RockMqMesgManage manage, Class<? extends RocketMqMessageStandardOprate> oprate,
			int threadNum, ExecuteModel model) {
		this(single_key, config, manage, threadNum, model);
		this.oprate = oprate;
		type = ExecuteType.Class;
	}
	
	public RocketMesgExecute(String single_key, RocketMqConfig config, RockMqMesgManage manage, Class<? extends RocketMqMessageStandardOprate> oprate,
			int threadNum) {
		this(single_key, config, manage, threadNum, null);
		this.oprate = oprate;
		type = ExecuteType.Class;
	}
	
	public RocketMesgExecute(String single_key, RocketMqConfig config, RockMqMesgManage manage, RocketMqMessageStandardOprate r_oprate,
			int threadNum, ExecuteModel model) {
		this(single_key, config, manage, threadNum, model);
		this.r_oprate = r_oprate;
		type = ExecuteType.Example;
	}
	
	public RocketMesgExecute(String single_key, RocketMqConfig config, RockMqMesgManage manage, RocketMqMessageStandardOprate r_oprate,
			int threadNum) {
		this(single_key, config, manage, threadNum, null);
		this.r_oprate = r_oprate;
		type = ExecuteType.Example;
	}
	
	public void init() throws Exception{
		boolean status = false;
		switch(model){
		case Blocking_queue:
			threadManage = new ThreadExecuteManager(threadNum);
			status = ConsumerCurrencyCommon.openConsumerBlock(config, manage);
			break;
		case Nonblocking_queue:
			threadManage = new ThreadManage(threadNum);
			status = ConsumerCurrencyCommon.openConsumer(config, manage);
			break;
		}
		if(threadManage == null || !status){
			logger.error("任务:{}初始化执行任务失败", single_key);
			throw new Exception("任务:" + single_key + "初始化执行任务失败");
		}
		init_type = true;
	}

	public void start(){
		switch(model){
		case Blocking_queue:
			start_Block();
			break;
		case Nonblocking_queue:
			start_NoBlock();
			break;
		}
	}
	
	public void start_Block(){
		start_Block_Policy1();
//		start_Block_Policy2();
	}
	
	public void start_Block_Policy2(){
		if(executing){
			return;
		}
		for(int i = 0 ; i < threadNum ; i ++){
			TaskExecuteThread thread = new TaskExecuteThread(manage);
			threadManage.execute(thread);
		}
		executing = true;
	}
	
	public void start_Block_Policy1(){
		if(executing)return;
		long start = System.currentTimeMillis();
		long start_c = start;
		long end = start;
		long get_task_end = end;
		long init_task_end = end;
		int all_num = 0;
		int current_num = 0;
		while(true){
			long start_task = System.currentTimeMillis();
			executing = true;
			MesgTaskManage task = manage.getMesgTaskManage();
			if(task == null){
				logger.info("任务组：{}没有需要执行的任务", single_key);
				break;
			}
			get_task_end = System.currentTimeMillis();
			if(task.getTask_data() == null)continue;
			try{
				RocketMqMessageStandardOprate oprate = getRocketMqMessageStandardOprate(task);
				threadManage.execute(oprate);
			}catch(Exception e){
				manage.removeMesg(task.getMesg_key());
				logger.error("任务组：{}初始化任务异常", single_key, e);
			}
			init_task_end = System.currentTimeMillis();
			logger.info("获取任务耗时:{}, 初始化任务耗时:{}", (get_task_end - start_task), (init_task_end - start_task));
			all_num ++;
			current_num ++;
			end = System.currentTimeMillis();
			if(end - start_c > 1 * 60 * 1000){
				logger.info("single_key:{},all_num:{},current_num:{},all_speed:{},avg_speed:{},thread_num:{},thread_ing_num:{},queue_length:{},mesg_cache:{}"
						, single_key, all_num, current_num, all_num/((end - start)/1000), current_num/((end - start_c)/1000), threadManage.getThread_num()
						,threadManage.getThreadNum(), manage.getQueueLength(), manage.getMesgNum());
				start_c = end;
				current_num = 0;
			}
		}
		executing = false;
	}
	
	public void start_NoBlock(){
		List<MesgTaskManage> tasks = manage.getMesgTaskManageAndUpdateStatus(true);
		logger.info("任务组：{}需要执行的任务数:{}", single_key, tasks.size());
		for(MesgTaskManage task : tasks){
			try{
				RocketMqMessageStandardOprate oprate_obj = getRocketMqMessageStandardOprate(task);
				threadManage.execute(oprate_obj);
			}catch(Exception e){
				manage.updateMessageStatus(task.getMesg_key(), false);
				logger.error("任务组：{}初始化任务异常", single_key, e);
			}
		}
	}
	
	public RocketMqMessageStandardOprate getRocketMqMessageStandardOprate(MesgTaskManage task) throws InstantiationException, IllegalAccessException, IllegalArgumentException, InvocationTargetException, NoSuchMethodException, SecurityException, CloneNotSupportedException{
		RocketMqMessageStandardOprate oprate_obj = null;
		switch(type){
		case Class:
			oprate_obj = (RocketMqMessageStandardOprate)
			oprate.getConstructor(RocketMqMessageStandard.class, RockMqMesgManage.class)
			.newInstance(task.getTask_data(), manage);
			break;
		case Example:
			oprate_obj = (RocketMqMessageStandardOprate)r_oprate.clone();
			oprate_obj.setMesg(task.getTask_data());
			oprate_obj.setManage(manage);
			break;
		default:
			break;
		}
		oprate_obj.setThreadmanager(threadManage);
		return oprate_obj;
	}

	public RocketMqConfig getConfig() {
		return config;
	}

	public void setConfig(RocketMqConfig config) {
		this.config = config;
	}

	public ThreadManager getThreadManage() {
		return threadManage;
	}

	public void setThreadManage(ThreadManager threadManage) {
		this.threadManage = threadManage;
	}

	public RocketMqMessageStandardOprate getR_oprate() {
		return r_oprate;
	}

	public void setR_oprate(RocketMqMessageStandardOprate r_oprate) {
		this.r_oprate = r_oprate;
	}

	public ExecuteModel getModel() {
		return model;
	}

	public void setModel(ExecuteModel model) {
		this.model = model;
	}

	public RockMqMesgManage getManage() {
		return manage;
	}

	public void setManage(RockMqMesgManage manage) {
		this.manage = manage;
	}

	public String getSingle_key() {
		return single_key;
	}

	public void setSingle_key(String single_key) {
		this.single_key = single_key;
	}

	public Class<? extends RocketMqMessageStandardOprate> getOprate() {
		return oprate;
	}

	public void setOprate(Class<? extends RocketMqMessageStandardOprate> oprate) {
		this.oprate = oprate;
	}

	public ExecuteType getType() {
		return type;
	}

	public void setType(ExecuteType type) {
		this.type = type;
	}

	public int getThreadNum() {
		return threadNum;
	}

	public boolean isInit_type() {
		return init_type;
	}

	public void setInit_type(boolean init_type) {
		this.init_type = init_type;
	}

	public void setThreadNum(int threadNum) {
		this.threadNum = threadNum;
	}
	
	public enum ExecuteType{
		Example,Class
	}
	
	public enum ExecuteModel{
		Blocking_queue,Nonblocking_queue
	}
	
	public class TaskExecuteThread implements Runnable{
		
		private RockMqMesgManage manage;
		
		private volatile boolean execute = true;
		
		private long start = System.currentTimeMillis();
		
		private long end = System.currentTimeMillis();
		
		private int execute_num = 0;
		
		public TaskExecuteThread(RockMqMesgManage manage) {
			super();
			this.manage = manage;
		}

		public RockMqMesgManage getManage() {
			return manage;
		}

		public void setManage(RockMqMesgManage manage) {
			this.manage = manage;
		}

		public boolean isExecute() {
			return execute;
		}

		public void setExecute(boolean execute) {
			this.execute = execute;
		}

		@Override
		public void run() {
			// TODO Auto-generated method stub
			while(execute){
				try{
					MesgTaskManage task = manage.getMesgTaskManage();
					if(task == null){
						Thread.sleep(10 * 1000);
						break;
					}
					if(task.getTask_data() == null)continue;
					RocketMqMessageStandardOprate oprate = getRocketMqMessageStandardOprate(task);
					oprate.execute(false);
					manage.removeMesg(task.getMesg_key());
					execute_num ++;
					end = System.currentTimeMillis();
					if(end - start > 60 * 1000){
						logger.info("thread_name:{},execute_num:{}", Thread.currentThread().getName(), execute_num);
						execute_num = 0;
						start = end;
					}
				}catch(Exception e){
					logger.error("");
				}
			}
		}
	}
}
